package com.zy.gmall.realtime.app.dwd.log;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.zy.gmall.realtime.util.DateFormatUtil;
import com.zy.gmall.realtime.util.KafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class DwdTrafficBaseLogSplit {
    public static void main(String[] args) {
        //1 创建环境 设置并行度
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //2 设置检查点 检查点相关配置
        //3 获取kafka 主题 数据源

        String topic =  "topic_log";
        String groupId = "test";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic ,groupId);
        DataStreamSource<String> kafkaSourceDs = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafkaSource");

        //kafkaSourceDs.print("111:::::");

        OutputTag<String> blackTag = new OutputTag<String>("black"){};
        //4 数据转换，并且进行数据清洗
        SingleOutputStreamOperator<JSONObject> outputStreamOperator = kafkaSourceDs.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                //判断是否是合格的JSON
                try {
                    JSONObject jsonObject = JSON.parseObject(s);
                    collector.collect(jsonObject);
                    //System.out.println("发送到下游");
                }catch (Exception e){
                    // 如果出错不合格 就进入侧流
                    context.output(blackTag, s);
                    //System.out.println("进侧流");
                }
            }
        });
        //5 输出 格式不正确的 侧流
        SideOutputDataStream<String> sideOutput = outputStreamOperator.getSideOutput(blackTag);
        //sideOutput.print("!!!!!!!");
        //输出到kafka topic主题中
        sideOutput.sinkTo(KafkaUtil.getKafkaSink("topic_black"));

        //6 分区 每个用户有自己的状态
        KeyedStream<JSONObject,String> keyedStream = outputStreamOperator.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getJSONObject("common").getString("mid");
            }
        });

        //7 进行新老用户状态标记回复 注意时间
        SingleOutputStreamOperator<JSONObject> processDs = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                    //使用状态,存储keyBy之后的数据的第一次登陆时间
                    private ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //初始化状态
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("valueState", String.class);
                        valueState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public void processElement(JSONObject jsonObject, Context context, Collector<JSONObject> collector) throws Exception {
                        //获取状态中日期
                        String state = valueState.value();
                        //获取当前日期
                        Long ts = jsonObject.getLong("ts");
                        //时间格式转换
                        String toDate = DateFormatUtil.toDate(ts);
                        //获取当前数据 is_new
                        String is_new = jsonObject.getJSONObject("common").getString("is_new");
                        //判断
                        if ("1".equals(is_new)) {
                            //为1 判断状态是否有 没有就加
                            if (StringUtils.isEmpty(state)) {
                                valueState.update(toDate);
                            } else {
                                //有的话判断是否一致
                                if (!state.equals(toDate)) {
                                    //不一致的话
                                    is_new = "0";
                                    jsonObject.getJSONObject("common").put("is_new", is_new);
                                }
                            }
                        } else {
                            //不为1
                            //判断状态中是否有
                            if (StringUtils.isEmpty(state)) {
                                //为空的话，加上
                                valueState.update(DateFormatUtil.toDate(ts - 24 * 60 * 60 * 1000));
                            }

                        }
                        collector.collect(jsonObject);
                    }
                }
        );
        //processDs.print();

        //定义侧流
        OutputTag<String> errorTag = new OutputTag<>("errorTag",TypeInformation.of(String.class));
        OutputTag<String> startTag = new OutputTag<String>("startTag"){};
        OutputTag<String> displayTag = new OutputTag<>("displaysTag",TypeInformation.of(String.class));
        OutputTag<String> actionTag = new OutputTag<>("actionsTag",TypeInformation.of(String.class));

        //8 分流
        SingleOutputStreamOperator<String> pageDs = processDs.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, String>.Context context, Collector<String> collector) throws Exception {
                        //判断是否存在error
                        JSONObject errJsonObj = jsonObject.getJSONObject("err");
                        if (errJsonObj != null) {
                            //不等于空，有错
                            //放入侧流，只放err太少了
                            context.output(errorTag, jsonObject.toJSONString());
                            jsonObject.remove("err");
                        }
                        //非报错，正常启动数据日志
                        JSONObject startJsonObj = jsonObject.getJSONObject("start");
                        if (startJsonObj != null) {
                            //放入侧流
                            context.output(startTag, jsonObject.toJSONString());
                        } else {
                            //启动日志为空，则可能是页面日志 页面日志包括 common ts page
                            Long ts = jsonObject.getLong("ts");
                            JSONObject page = jsonObject.getJSONObject("page");
                            JSONObject common = jsonObject.getJSONObject("common");

                            //曝光日志
                            JSONArray display = jsonObject.getJSONArray("displays");
                            if (display != null && display.size() > 0) {

                                for (int i = 0; i < display.size(); i++) {
                                    JSONObject displayJSONObject = display.getJSONObject(i);
                                    JSONObject object = new JSONObject();
                                    object.put("display", displayJSONObject);
                                    object.put("ts", ts);
                                    object.put("page", page);
                                    object.put("common", common);
                                    context.output(displayTag, object.toJSONString());
                                }
                                jsonObject.remove("displays");
                            }
                            //动作日志

                            JSONArray actions = jsonObject.getJSONArray("actions");
                            if (actions != null && actions.size() > 0) {
                                //存在动作日志
                                for (int i = 0; i < actions.size(); i++) {
                                    //获得array中每一个动作日志
                                    JSONObject actionsJSONObject = actions.getJSONObject(i);
                                    JSONObject object = new JSONObject();
                                    object.put("action", actionsJSONObject);
                                    object.put("common", common);
                                    object.put("page", page);
                                    context.output(actionTag, object.toJSONString());
                                }
                                jsonObject.remove("actions");
                            }
                        }
                        collector.collect(jsonObject.toJSONString());
                    }
                }
        );
        SideOutputDataStream<String> errDs = pageDs.getSideOutput(errorTag);
        SideOutputDataStream<String> startDs = pageDs.getSideOutput(startTag);
        SideOutputDataStream<String> displayDs = pageDs.getSideOutput(displayTag);
        SideOutputDataStream<String> actionDs = pageDs.getSideOutput(actionTag);

        errDs.print("!!!!!!");
        startDs.print("$$$$$");
        displayDs.print("%%%%%%");
        actionDs.print("******");
        //9 输出
        errDs.sinkTo(KafkaUtil.getKafkaSink("topic_err"));
        startDs.sinkTo(KafkaUtil.getKafkaSink("topic_start"));
        displayDs.sinkTo(KafkaUtil.getKafkaSink("topic_displays"));
        actionDs.sinkTo(KafkaUtil.getKafkaSink("topic_actions"));

        //{"common":{"ar":"13","uid":"46","os":"Android 13.0","ch":"wandoujia","is_new":"1","md":"OPPO Remo8","mid":"mid_498","vc":"v2.1.134","ba":"OPPO","sid":"bf03677e-8a09-45eb-9caa-7dd2e6a80ff9"},
        // "page":{"page_id":"trade","item":"14,19,20,1,4","during_time":12433,"item_type":"sku_ids","last_page_id":"cart"},
        // "ts":1654771816000}
        SingleOutputStreamOperator<String> page = pageDs.process(new ProcessFunction<String, String>() {
            @Override
            public void processElement(String s, ProcessFunction<String, String>.Context context, Collector<String> collector) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(s);
                if (jsonObject.getJSONObject("page") != null) {
                    collector.collect(JSON.toJSONString(jsonObject));
                }
            }
        });
        //page.print(">>>>>");
        page.sinkTo(KafkaUtil.getKafkaSink("topic_pages"));

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
